package org.sg20.parsename;

import org.ansj.domain.Result;
import org.ansj.domain.Term;
import org.ansj.library.DicLibrary;
import org.ansj.splitWord.analysis.DicAnalysis;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class ParseNameWithLib {
    public static class ParseNameWithLibMapper extends Mapper<LongWritable, Text, ParseNameOutputKey, Text> {
        ParseNameOutputKey outkey;

        @Override
        protected void setup(Mapper<LongWritable, Text, ParseNameOutputKey, Text>.Context context) throws IOException {
            Path nameList = new Path(context.getCacheFiles()[0]);
            FileSystem fs = FileSystem.get(context.getConfiguration());
            InputStreamReader isr = new InputStreamReader(fs.open(nameList));

            BufferedReader in = new BufferedReader(isr);
            // read until end of file
            String name;
            while ((name = in.readLine()) != null) {
                // add the name to the trie tree
                DicLibrary.insert(DicLibrary.DEFAULT, name);
            }
            in.close();
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ParseNameOutputKey, Text>.Context context) throws IOException, InterruptedException {
            String filename = ((FileSplit) context.getInputSplit()).getPath().getName();

            outkey = new ParseNameOutputKey(filename, key.get());
            StringBuilder str = new StringBuilder();

            Result terms = DicAnalysis.parse(value.toString());
            for (Term term : terms) {
                if (term.getNatureStr().equals("userDefine")) {
                    str.append(term.getName());
                    str.append("\t");
                }
            }
            if (!str.toString().equals(""))
                context.write(outkey, new Text(str.toString()));
        }
    }

    public static class ParseNameWithLibReducer extends Reducer<ParseNameOutputKey, Text, ParseNameOutputKey, Text> {
        // There is nothing to do for reducer
        @Override
        protected void reduce(ParseNameOutputKey key, Iterable<Text> values, Reducer<ParseNameOutputKey, Text, ParseNameOutputKey, Text>.Context context) throws IOException, InterruptedException {
            super.reduce(key, values, context);
        }
    }

}
